#!/bin/bash
set -euo pipefail

# 健康检查只检查端口
check_kafka_ready() {
  while ! nc -z localhost 9094; do
    echo "Waiting for Kafka to start..."
    sleep 5
  done
  echo "Kafka is up on port 9094"
}

# 主题创建函数
create_topics() {
  if [ -n "$KAFKA_CREATE_TOPICS" ]; then
    IFS=',' read -ra TOPICS <<< "$KAFKA_CREATE_TOPICS"
    for topic in "${TOPICS[@]}"; do
      IFS=':' read -ra PARTS <<< "$topic"
      
      # 参数校验
      if [[ "${#PARTS[@]}" -lt 3 ]]; then
        echo "ERROR: Invalid topic format '${topic}'. Expected: name:partitions:replicas" >&2
        continue
      fi

      # 重试逻辑
      local retries=3
      local delay=5
      for ((i=1; i<=retries; i++)); do
        if /opt/kafka/bin/kafka-topics.sh --create \
          --bootstrap-server localhost:9094 \
          --topic "${PARTS[0]}" \
          --partitions "${PARTS[1]:-1}" \
          --replication-factor "${PARTS[2]:-1}"; then
          echo "SUCCESS: Created topic ${PARTS[0]}"
          break
        else
          echo "WARNING: Attempt $i failed to create topic ${PARTS[0]}" >&2
          if [ $i -lt $retries ]; then
            sleep $delay
          else
            echo "ERROR: Failed to create topic ${PARTS[0]} after $retries attempts" >&2
          fi
        fi
      done
    done
  fi
}

main() {
  check_kafka_ready
  create_topics
}

main "$@"